package org.apache.rocketmq.test.app.producer;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.List;

/**
 * 测试：选择消息队列发送消息
 * 发送消息时，使用 MessageQueueSelector(消息队列选择器) 实现将消息指定发送到某个队列，可以基于这个特点，实现发送局部顺序消息
 *
 * @author qinyupeng
 * @since 2021-12-29 15:25:04
 */
public class SendOrderedMsg {

    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("QS_PRODUCER_GROUP");
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.start();

        String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD"};
        String topic = "QS_ORDERED_TOPIC";
        for (int i = 0; i < 10; i++) {
            int orderId = i % 2;
            String tagName = tags[i % tags.length];
            System.out.println("tagName=" + tagName);
            Message message = new Message(topic, tagName, "KEY" + i, ("订单号：" + orderId).getBytes());
            // 设置property，消费端可以使用sql92模式从broker端拉取数据
            if ("TagB".equals(tagName)) {
                message.putUserProperty("orderStatus", "1");
            } else {
                message.putUserProperty("orderStatus", "0");
            }

            producer.send(message, new MessageQueueSelector() {
                // mqs为topic下所有消息队列
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    // 这里的id即为orderId
                    Integer id = (Integer) arg;
                    // 相同的orderId取模后得到的index值是相同的，也就是说，最终会将orderId相同的消息发送到同一个消息队列中
                    int index = id % mqs.size();
                    System.out.println("消息发送到 index=" + index);
                    return mqs.get(index);
                }
            }, orderId);

        }

        producer.shutdown();
    }

}
